package v1

import (
	"context"
	"encoding/json"
	"sync"
	"time"

	"hermestrans/basis/middleware"
	"hermestrans/basis/utils"
	"hermestrans/internal/mq"
	config "hermestrans/internal/portal/config"
	mgr "hermestrans/internal/portal/manager"
	"hermestrans/internal/portal/schema"

	"github.com/IBM/sarama"
	"github.com/gin-gonic/gin"
	"github.com/sirupsen/logrus"
	"go.mongodb.org/mongo-driver/internal/uuid"
)

type ApiHandler struct {
	conf          *config.Config
	dbMgr         *mgr.MgoMgr
	asyncProducer *mq.Producer
}

var (
	handlerInstance *ApiHandler
	handlerOnce     sync.Once
)

func NewApiHandler(conf *config.Config,
	dbMgr *mgr.MgoMgr) *ApiHandler {
	handlerOnce.Do(func() {
		asyncProducer := mq.NewKafkaProducer(conf.KfkConfig.Brokers, conf.KfkConfig.User, conf.KfkConfig.Passwd)
		handlerInstance = &ApiHandler{
			conf:          conf,
			dbMgr:         dbMgr,
			asyncProducer: asyncProducer,
		}
	})
	return handlerInstance
}

// 新增
func (a *ApiHandler) HandleJobAdd(c *gin.Context) {
	var (
		err     error
		jobInfo *schema.Job
		byteJob []byte
	)
	//参数检测
	if err = c.ShouldBindJSON(&jobInfo); err != nil {
		middleware.ErrResponse(c, utils.ParamTypeError, err.Error())
	}

	now := time.Now()
	jobId := "J" + uuid.NewV4()
	job := &schema.Job{
		Id:              jobId,                //作业的唯一标识
		CreaterId:       "",                   //todo 暂时留空，等用户系统完善
		Creater:         "",                   //todo 暂时留空，等用户系统完善
		UpdaterId:       "",                   //todo 暂时留空，等用户系统完善
		Updater:         "",                   //todo 暂时留空，等用户系统完善
		CreateDate:      now,                  //
		UpdateDate:      now,                  //
		Status:          "1",                  //
		Code:            jobId,                //
		Name:            jobInfo.Name,         //作业名称
		JobType:         jobInfo.JobType,      //作业类型
		TransferType:    jobInfo.TransferType, //传输方式
		FromView:        jobInfo.FromView,     //来源
		IsRegular:       jobInfo.IsRegular,    //正则表达式匹配时传输
		IsRecursive:     jobInfo.IsRecursive,  //遍历子目录
		IsEncrypt:       jobInfo.IsEncrypt,    //是否加密
		Describle:       jobInfo.Describle,    //作业描述
		SrcNodeId:       jobInfo.SrcNodeId,    //源节点名称
		SrcFilePath:     jobInfo.SrcFilePath,  //源文件路径
		SrcFileName:     jobInfo.SrcFileName,  //源文件表达式
		SrcFlagFileName: jobInfo.SrcFileName,  //标志文件名
		DestNodeId:      jobInfo.DestNodeId,   //目标名称
		DestFilePath:    jobInfo.DestFilePath, //保存路径
	}
	jobs := []*schema.Job{job}
	if err = a.dbMgr.EditJobs(jobs); err != nil {
		middleware.ErrResponse(c, utils.DBEditError, err.Error())
	}
	if byteJob, err = json.Marshal(job); err != nil {
		middleware.ErrResponse(c, utils.DBEditError, err.Error())
	}
	//发送到mq里进行
	a.asyncProduceMsg(a.conf.KfkConfig.Topic, string(byteJob), a.conf.KfkConfig.Enqueued,
		a.conf.KfkConfig.Timeout, a.conf.KfkConfig.Errs)
	return
}

func (a *ApiHandler) asyncProduceMsg(topic, fullInfo string, enqueued, timeout, errors int) {
	msg := &sarama.ProducerMessage{Topic: topic, Key: nil, Value: sarama.StringEncoder(fullInfo)}
	producer := a.asyncProducer.AsyncProducer
	// 异步发送只是写入内存了就返回了，并没有真正发送出去
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*60000)

	select {
	case producer.Input() <- msg:
		enqueued++
	case err := <-producer.Errors():
		errors++
		logrus.Error(err.Error(), " fail to produce message . errors: ", errors)
	case <-ctx.Done():
		timeout++
	}

	cancel()
}

// 修改
func (a *ApiHandler) HandleJobsEdit(c *gin.Context) {
	var (
		err  error
		jobs []*schema.Job
	)
	//参数检测
	if err = c.ShouldBindJSON(&jobs); err != nil {
		middleware.ErrResponse(c, utils.ParamTypeError, err.Error())
	}
	if err = a.dbMgr.EditJobs(jobs); err != nil {
		middleware.ErrResponse(c, utils.DBEditError, err.Error())
	}
	//todo推送到mq中

}

// 删除
func (a *ApiHandler) HandleJobsDelete(c *gin.Context) {
	var (
		err  error
		jobs []*schema.Job
	)
	//参数检测
	if err = c.ShouldBindJSON(&jobs); err != nil {
		middleware.ErrResponse(c, utils.ParamTypeError, err.Error())
	}
	if err = a.dbMgr.EditJobs(jobs); err != nil {
		middleware.ErrResponse(c, utils.DBEditError, err.Error())
	}
	//todo推送到mq中

}

// 查询
func (a *ApiHandler) HandleJobsGet(c *gin.Context) {

}
